package flink.ftp;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by liujian on 2021/6/28.
 */
public class FtpDataSource extends RichParallelSourceFunction<String> {

    private static final String SP = "/";
    private BufferedReader br;

    private FTPClient ftpClient = null;


    public FTPClient getFtpClient() {
        return ftpClient;
    }

    @Override
    public void open(Configuration ftpConfig) throws Exception {
        ftpClient = new FTPClient();
        ftpClient.connect("192.168.140.146", 21);
//        ftpClient.login("dkx", "Pe*Th92Ajnjf8UY2");
        ftpClient.login("dkx", "dkx");
        ftpClient.setControlEncoding("UTF-8");
//        ftpClient.setConnectTimeout();
//        ftpClient.setDataTimeout();


    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        ftpClient.enterRemotePassiveMode();
        ftpClient.enterLocalPassiveMode();
        List<String> strings = listDirs("2021-06-04/");
        for (String path : strings) {
            System.out.println(path);
            InputStream in = getInputStream(path);
            br = new BufferedReader(new InputStreamReader(in, "GBK"));
            String line = null;
            while ((line = br.readLine()) != null) {
//                System.out.println(line);
                if (line.contains("XMPP--->请求参数requestService")) {
//                    System.out.println(line);
                    String jsonStr = line.replaceAll("XMPP--->请求参数requestService：", "").replaceAll(";", "");
                    JSONObject json = JSON.parseObject(jsonStr);
                    JSONArray dataItems = json.getJSONArray("dataItems");
                    for (Object dataItem : dataItems) {
                        JSONObject jsonObject = (JSONObject) JSONObject.toJSON(dataItem);
                        JSONObject tmp = jsonObject.getJSONObject("attributes");
                        Gson gson = new Gson();
                        AttributesInfo attributesInfo = gson.fromJson(tmp.toJSONString(), AttributesInfo.class);
                        ctx.collect(JSON.toJSONString(attributesInfo));
                    }
                }
            }
        }
    }

    @Override
    public void cancel() {

    }


    private List<String> listDirs(String path) {
        List<String> sources = new ArrayList<>();
        if (!path.endsWith(SP)) {
            path = path + SP;
        }
        try {
            FTPFile[] ftpFiles = ftpClient.listFiles(new String(path.getBytes(StandardCharsets.UTF_8), FTP.DEFAULT_CONTROL_ENCODING));
            if (ftpFiles != null) {
                for (FTPFile ftpFile : ftpFiles) {
                    if (StringUtils.endsWith(ftpFile.getName(), ConstantValue.POINT_SYMBOL) || StringUtils.endsWith(ftpFile.getName(), ConstantValue.TWO_POINT_SYMBOL)) {
                        continue;
                    }
                    sources.add(path + ftpFile.getName());
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        return sources;
    }

    public InputStream getInputStream(String filePath) {
        try {
            ftpClient.enterLocalPassiveMode();
            InputStream is = ftpClient.retrieveFileStream(new String(filePath.getBytes(StandardCharsets.UTF_8), FTP.DEFAULT_CONTROL_ENCODING));
            return is;
        } catch (IOException e) {
            String message = String.format("读取文件 : [%s] 时出错,请确认文件：[%s]存在且配置的用户有权限读取", filePath, filePath);
            throw new RuntimeException(message, e);
        }
    }


}
